package com.feidee.fd.sml.algorithm.feature

import com.feidee.fd.sml.algorithm.component.feature.{NormalizeEncoder, NormalizeEncoderParam}
import com.feidee.fd.sml.algorithm.util.{TestingDataGenerator, ToolClass}
import org.apache.spark.ml.linalg.Vectors
import org.scalatest.FunSuite

/**
  * @Author tangjinyuan
  * @Date 2019/3/28 14:32
  * @Description
  * @Reviewer
  */
class NormalizeEncoderSuite extends FunSuite {
  val paramStr: String =
    """
      |{
      |	"inputCol": "features",
      |	"outputCol": "normFeatures",
      | "p":1
      |}
    """.stripMargin

  val model = new NormalizeEncoder()
  val param: NormalizeEncoderParam = model.parseParam(new ToolClass().encrypt(paramStr))

  test("model parameter") {
    assert("features".equals(param.inputCol) && "normFeatures".equals(param.outputCol) && param.p == 1.0)
  }


  test("model transformation") {

    val ComparaDataFrame = TestingDataGenerator.spark.createDataFrame(Seq(
      (0, Vectors.dense(0.4, 0.2, -0.4)),
        (0, Vectors.dense(0.5, 0.25, 0.25)),
        (0, Vectors.dense(0.25, 0.625, 0.125))

    ))
      .toDF("id", "normFeatures")
    println("compareData:")
    ComparaDataFrame.select("normFeatures").show()

    val dataFrame = TestingDataGenerator.spark.createDataFrame(Seq(
      (0, Vectors.dense(1.0, 0.5, -1.0)),
      (1, Vectors.dense(2.0, 1.0, 1.0)),
      (2, Vectors.dense(4.0, 10.0, 2.0))
    )).toDF("id", "features")


    val res = model.train(param, dataFrame).transform(dataFrame)
    println("res:")
    res.select("normFeatures").show()

    assert(ComparaDataFrame.select("normFeatures").except(res.select("normFeatures")).count() == 0)
  }


}

